﻿
namespace Apache.Cassandra.Client {
    using System;
    using System.Collections.Generic;
    using System.Diagnostics;
    using System.Net;
    using System.Reflection;
    using Apache.Cassandra.Client.Diagnostics;
    using Thrift.Protocol;
    using Thrift.Transport;

    internal class CassandraConnectionFactory {

        internal static readonly string PerformanceCountersInstanceName = GetPerformanceCountersInstanceName();

        private IDictionary<object, CassandraConnectionPool> pools;
        private CassandraClientFactoryCounters performanceCounters;

        private CassandraConnectionFactory() {
            pools = new Dictionary<object, CassandraConnectionPool>();
            performanceCounters = new CassandraClientFactoryCounters(PerformanceCountersInstanceName);
        }

        private static readonly CassandraConnectionFactory singleton = new CassandraConnectionFactory();
        public static CassandraConnectionFactory Factory {
            get {
                return CassandraConnectionFactory.singleton;
            }
        }

        internal CassandraClientFactoryCounters PerformanceCounters {
            get {
                return performanceCounters;
            }
        }

        internal CassandraInternalConnection GetConnection(CassandraConnection owner) {
            Debug.Assert(null != owner);

            CassandraInternalConnection connection;
            CassandraConnectionPool pool = GetClusterPool(owner);
            if (null == pool) {
                CassandraConnectionString options = owner.Options;
                connection = CreateConnection(null, options);
                if (null == connection) {
                    throw new InvalidOperationException();
                }

                CassandraConnectionFactory.Factory.PerformanceCounters.HardConnectsPerSecond.Increment();
                connection.MakeNonPooledConnection(owner);
                CassandraConnectionFactory.Factory.PerformanceCounters.NumberOfNonPooledConnections.Increment();
            } else {
                connection = pool.PopConnection(owner);
            }

            if (null == connection) {
                // TODO: Unable to connect
                throw new Exception();
            }

            return connection;
        }

        private CassandraConnectionPool GetClusterPool(CassandraConnection connection) {
            Debug.Assert(null != connection);
            Debug.Assert(null != connection.Options);

            CassandraConnectionString options = connection.Options;
            if (!options.Pooling) {
                return null;
            }

            object connectionString = options.ConnectionString;
            Debug.Assert(null != connectionString);

            CassandraConnectionPool pool;
            if (!pools.TryGetValue(connectionString, out pool)) {
                lock (pools) {
                    if (!pools.TryGetValue(connectionString, out pool)) {
                        IDictionary<object, CassandraConnectionPool> newpools = new Dictionary<object, CassandraConnectionPool>(pools.Count + 1);
                        foreach (var i in pools) {
                            newpools.Add(i);
                        }

                        pool = new CassandraConnectionPool(options);
                        pool.Startup();

                        newpools.Add(connectionString, pool);
                        pools = newpools;

                        PerformanceCounters.NumberOfActiveConnectionPools.Increment();
                    }
                }
            }

            return pool;
        }

        internal CassandraInternalConnection CreateConnection(CassandraConnectionPool pool, CassandraConnectionString options) {
            EndPoint endpoint = new DnsEndPoint(options.Server, options.Port);
            return CreateConnection(pool, endpoint, options);
        }

        internal CassandraInternalConnection CreateConnection(CassandraConnectionPool pool, EndPoint endpoint, CassandraConnectionString options) {
            TTransport transport = CreateTransport(endpoint);
            if (null == transport) {
                return null;
            }

            TProtocol protocol = CreateProtocol(transport);
            if (null == protocol) {
                return null;
            }

            Cassandra.Client client = new Cassandra.Client(protocol);

            return new CassandraInternalConnection(client, endpoint, options);
        }

        private TTransport CreateTransport(EndPoint endpoint) {
            IPEndPoint ip = endpoint as IPEndPoint;
            DnsEndPoint dns = null == ip ? endpoint as DnsEndPoint : new DnsEndPoint(ip.Address.ToString(), ip.Port);
            if (null == dns) {
                return null;
            }

            TSocket socket = new TSocket(dns.Host, dns.Port);
            socket.TcpClient.NoDelay = true;
            return new TFramedTransport(socket);
        }

        private TProtocol CreateProtocol(TTransport transport) {
            return new TBinaryProtocol(transport);
        }

        static private string GetPerformanceCountersInstanceName() {
            // Get process name
            string processName = null;
            Assembly assembly = Assembly.GetEntryAssembly();
            if (null != assembly) {
                AssemblyName name = assembly.GetName();
                if (null != name) {
                    processName = name.Name;
                }
            }
            if (null == processName) {
                AppDomain appDomain = AppDomain.CurrentDomain;
                if (null != appDomain) {
                    processName = appDomain.FriendlyName;
                }
            }

            // Get process id
            int pid = SafeNativeMethods.GetCurrentProcessId();

            // Create instance name
            string instanceName = String.Format((IFormatProvider)null, "{0}[{1}]", processName, pid);
            instanceName = instanceName.Replace('(','[').Replace(')',']').Replace('#','_').Replace('/','_').Replace('\\','_');

            return instanceName;
        }
    }
}
